﻿using Confluent.Kafka;
using KafkaNet;
using KafkaNet.Model;
using KafkaNet.Protocol;
using System;
using System.Collections.Generic;
using System.Text;

namespace Zhaoxi.KafkaClient
{
	class NetKafka
	{
		public static void Pull()
		{
			var options = new KafkaOptions(new Uri("http://127.0.0.1:9092"));
			var router = new BrokerRouter(options);
			var consumer = new Consumer(new ConsumerOptions("MyTopic", router));
			OffsetPosition p0 = new OffsetPosition(0, 100);
			OffsetPosition p2 = new OffsetPosition(0, 100);
			consumer.SetOffsetPosition(new OffsetPosition[] { p0,p2 });
			while (1 == 1)
			{
				var msgs = consumer.Consume();
				foreach (var msg in msgs)
				{
					Console.WriteLine(Encoding.UTF8.GetString(msg.Value));
				}
			}

		}
	}
}
